package cn.gwm.flink.streaming.dwd.function;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.json.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @ClassName ShuntStreamProcess
 * @Description
 * @Author LiangGuang
 * @Date 2023/05/18 15:49
 */
public class ShuntStreamProcess extends ProcessFunction<JSONObject, JSONObject> {

    private OutputTag<JSONObject> errorStream;

    public ShuntStreamProcess(OutputTag<JSONObject> errorStream) {
        this.errorStream = errorStream;
    }

    @Override
    public void processElement(JSONObject item, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        if (item.containsKey(BaseFields.ERROR)) {
            ctx.output(errorStream, item);
        } else {
            out.collect(item);
        }
    }
}
